package cloud.xiguapi.ubas.analysis.uv;

import cloud.xiguapi.ubas.analysis.uv.model.BloomFilterUvCountResultProcess;
import cloud.xiguapi.ubas.analysis.uv.model.UVTrigger;
import cloud.xiguapi.ubas.model.UserBehavior;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

import static cloud.xiguapi.ubas.constant.JobNameConstant.BLOOM_UV;

/**
 * @author 大大大西西瓜皮🍉
 * date: 2021-5-19 上午 09:52
 * desc:
 */
@Component
public class BloomUVAnalysis {

    @Resource(name = "env")
    private StreamExecutionEnvironment env;

    @Async("bloomUVAnalysis")
    public void analysis(String path) throws Exception {
        env.setParallelism(8);

        // 设置时间语义
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // 读取数据
        DataStream<String> inputDataStream = env.readTextFile(path);

        // 转换, 并分配时间戳和水印
        DataStream<UserBehavior> userBehaviorDataStream = inputDataStream.map(line -> {
            String[] fields = line.split(",");
            return UserBehavior.builder()
                    .userId(Long.valueOf(fields[0]))
                    .itemId(Long.valueOf(fields[1]))
                    .categoryId(Long.valueOf(fields[2]))
                    .behavior(fields[3])
                    .timestamp(Long.valueOf(fields[4]))
                    .build();
        }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<UserBehavior>(
                Time.of(200, TimeUnit.MILLISECONDS)) {
            @Override
            public long extractTimestamp(UserBehavior element) {
                return element.getTimestamp() * 1000L;
            }
        });

        // 开窗聚合
        userBehaviorDataStream.filter(data -> "pv".equals(data.getBehavior()))
                .timeWindowAll(Time.hours(1L))
                .trigger(new UVTrigger())
                .process(new BloomFilterUvCountResultProcess());

        env.execute(BLOOM_UV.getJobName());
    }
}
